package net.htjs.flinkcdc.config;

import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import net.htjs.flinkcdc.process.TableDataProcessFunction;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.Properties;

@Configuration
public class FlinkCdcConfig {

    // 需要同步的表列表
    private static final String[] TABLE_LIST = {
        "fcyth_jiananqu.fcyth_jk_log"
    };

    @Bean
    public StreamExecutionEnvironment createExecutionEnvironment() throws Exception {
        // 1. 创建Flink执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 2. 配置检查点（根据需要调整）
        // 在Flink环境中增加检查点超时时间 // 60秒
        env.enableCheckpointing(60000);
        // 120秒超时
        env.getCheckpointConfig().setCheckpointTimeout(120000);
        // 配置其他检查点参数...

        // 3. 创建MySQL CDC源
        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
                .hostname("localhost")
                .port(3306)
                .username("root")
                .password("Tian@0628")
                // 配置需要同步的数据库和表
                .databaseList("fcyth_jiananqu")
                .tableList(TABLE_LIST)
                // 启动选项：从最新位置开始同步
                .startupOptions(StartupOptions.latest())
//                .startupOptions(StartupOptions.initial())
                // 配置序列化方式（JSON格式，包含表名等元数据）
                .deserializer(new JsonDebeziumDeserializationSchema())
                // 配置debezium属性（可选）
                .debeziumProperties(getDebeziumProperties())
                .build();

        // 4. 从CDC源创建数据流
        DataStream<String> stream = env.fromSource(
                mySqlSource,
                WatermarkStrategy.noWatermarks(),
                "MySQL CDC Source"
        );
        // 5. 处理数据流 - 根据表名分流处理
        stream.process(new TableDataProcessFunction());
//        stream.setParallelism(3);
        // 6. 执行Flink任务
        env.execute("Multi-table Sync with Flink CDC");

        return env;
    }

    // 配置Debezium属性
    private Properties getDebeziumProperties() {
        Properties properties = new Properties();
        // 增加这个配置可以在JSON中包含表名信息
        properties.setProperty("include.schema.changes", "false");
        properties.setProperty("database.server.name", "mysql-server");
        return properties;
    }
}
